//package com.ibm.cps.servlet;
//
//import java.util.ArrayList;
//import java.util.Collection;
//import java.util.List;
//import java.util.Map;
//
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//
//import com.fasterxml.jackson.databind.node.ArrayNode;
//import com.fasterxml.jackson.databind.node.ObjectNode;
//import com.ibm.cps.kafka.KafkaProducer;
//import com.ibm.cps.kafka.KafkaTopicConstructor;
//import com.ibm.cps.message.MessageFields;
//import com.ibm.cps.message.validation.ValidatorUtil;
//import com.ibm.cps.model.AdapaterFactory;
//import com.ibm.cps.model.DataSourceMonitor;
//import com.ibm.cps.model.MetadataAdapter;
//import com.ibm.cps.model.ServletUtil;
//import com.ibm.cps.newmessage.AbstractDataSourceMetadata;
//import com.ibm.cps.newmessage.DataSourceMetadataFactory;
//import com.ibm.factories.PublishProcessorFactory;
//import com.ibm.interfaces.IPersistency;
//import com.ibm.mongo.MongoPersistency;
//import com.ibm.util.ErrorCode;
//import com.ibm.util.JSONUtility;
//import com.ibm.util.LocalConfig;
//import com.ibm.util.exception.CPSException;
//
//public class DataSourceImplementation implements ServletTemplate {
//
//	private Logger logger;
//	private KafkaProducer producer;
//	private IPersistency persistency;
//	private DataSourceMonitor monitor;
//
//	public DataSourceImplementation() throws CPSException {
////		monitor = DataSourceMonitor.getInstance();
//		monitor = new DataSourceMonitor();
//		logger = LoggerFactory.getLogger(this.getClass());
//		producer = new KafkaProducer(LocalConfig.KAFKA_BROKERS);
//		persistency = MongoPersistency.getInstace();
//		logger.info("Connect to kafka: " + LocalConfig.KAFKA_BROKERS);
//		logger.info("Connect to mongo: " + LocalConfig.MONGO_CONNECT_STRING + ", and database is "
//				+ LocalConfig.PERSISTENCY_DATABASE);
//	}
//
//	@Override
//	public String doPost(Map<String, String> key2Value, String input) throws CPSException {
//		String tenantid = key2Value.get(MessageFields.TENANTID);
//		ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
//		String source = key2Value.get("source");
//		if (source == null) {
//			source = "rest";
//		}
//		return validateDataSource(tenantid, input);
//	}
//
//	@Override
//	public String doGet(Map<String, String> key2Value) throws CPSException {
//		String tenantid = key2Value.get(MessageFields.TENANTID);
//		ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
//		String source = key2Value.get(MessageFields.SOURCE);
//		if (source == null) {
//			source = "rest";
//		}
//
//		PublishProcessorFactory publishFactory = PublishProcessorFactory.getInstance();
//		MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantid), publishFactory, source);
//		Collection<String> metadatas = persistency.getTenantDataSource(tenantid, key2Value.get(MessageFields.NAME));
//		Collection<String> result = adapter.getJsons(metadatas, metadatas);
//		return ServletUtil.constructCollectionJson(result);
//	}
//
//	@Override
//	public String doDelete(Map<String, String> key2Value, String input) throws CPSException {
//		throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, "This method is not supported now.");
//	}
//
//	@Override
//	public String doDelete(Map<String, String> key2Value) throws CPSException {
//		String tenantid = key2Value.get(MessageFields.TENANTID);
//		ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
//		String source = key2Value.get(MessageFields.SOURCE);
//		if (source == null) {
//			source = "rest";
//		}
//		Collection<String> metadatas = persistency.getDataSource(tenantid, key2Value.get(MessageFields.NAME));
//		if (metadatas != null) {
//			for (String dataSource : metadatas) {
//				AbstractDataSourceMetadata metadata = DataSourceMetadataFactory.parseDataSourceMetadata(dataSource);
//				String name = metadata.getName();
//				ObjectNode node = JSONUtility.newObjectNode();
//				node.set(MessageFields.PARENTPROCESSORS, JSONUtility.newArrayNode().add(name));
//
//				Collection<String> inputs = persistency.getProcessors(node);
//				if (inputs != null && inputs.size() >= 1) {
//					throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, "Can't delete this data source "
//							+ metadata.getName() + " because it's referenced by other processor metadatas.");
//				}
//			}
//			List<String> deleted = new ArrayList<String>();
//
//			try {
//				for (String dataSource : metadatas) {
//					AbstractDataSourceMetadata metadata = DataSourceMetadataFactory.parseDataSourceMetadata(dataSource);
//					monitor.deleteDataSource(tenantid, metadata);
//					Collection<String> result = persistency.deleteDataSource(tenantid, metadata.getName());
//					producer.send(KafkaTopicConstructor.getDataSourceDeleteTopic(tenantid),
//							metadata.getInput().getBytes());
//
//					if (result != null) {
//						for (String ds : result) {
//							deleted.add(ds);
//						}
//					}
//				}
//			} catch (Throwable e) {
//
//				// transactionality
//				for (String dataSource : deleted) {
//					monitor.validateDataSource(tenantid, JSONUtility.fromObjectString(dataSource));
//					persistency.putDataSource(dataSource);
//					producer.send(KafkaTopicConstructor.getDataSourceAddTopic(tenantid), dataSource.getBytes());
//				}
//				throw e;
//			}
//			return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, ServletUtil.constructCollectionJson(deleted));
//		}
//		return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing to be deleted");
//	}
//
//	@Override
//	public String doPut(Map<String, String> key2Value, String input) throws CPSException {
//		throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, "Put method for data source is unsupported.");
//	}
//
//	/**
//	 * validate the data source metadata and put it in persistency and send to
//	 * kafka
//	 *
//	 * @param tenantid
//	 * @param input
//	 * @return
//	 * @throws CPSException
//	 */
//	public String validateDataSource(String tenantid, String input) throws CPSException {
//		if (input == null || input.length() < 1) {
//			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "Nothing to be posted.");
//		}
//		ArrayList<AbstractDataSourceMetadata> sendObjects = new ArrayList<AbstractDataSourceMetadata>();
//		try {
//			ArrayNode array = JSONUtility.getArrayFromString(input);
//			ArrayNode retArray = JSONUtility.newArrayNode();
//
//			for (int i = 0; i < array.size(); i++) {
//				ObjectNode objectNode = (ObjectNode) array.get(i);
//				AbstractDataSourceMetadata dataSource = monitor.validateDataSource(tenantid, objectNode);
//				ObjectNode retNode = JSONUtility.newObjectNode();
//				retNode.put("id", dataSource.getName());
//				retArray.add(retNode);
//				sendObjects.add(dataSource);
//			}
//
//			for (int i = 0; i < sendObjects.size(); i++) {
//				persistency.putDataSource(sendObjects.get(i).getInput());
//				producer.send(KafkaTopicConstructor.getDataSourceAddTopic(tenantid),
//						sendObjects.get(i).getInput().getBytes());
//			}
//			return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, retArray);
//		} catch (CPSException e) {
//			for (int i = 0; i < sendObjects.size(); i++) {
//				AbstractDataSourceMetadata tmpObject = sendObjects.get(i);
//				persistency.deleteDataSource(tenantid, tmpObject.getName());
//				monitor.deleteDataSource(tenantid, tmpObject);
//			}
//			throw e;
//		}
//	}
//}
